package com.winning.platwebservice.service.Impl;

import com.winning.platwebservice.controller.KafkaCollect;
import com.winning.platwebservice.dao.DataProcessingDao;
import com.winning.platwebservice.service.CommonService;
import com.winning.platwebservice.utils.EhcacheUtil;
import com.winning.platwebservice.utils.ValidateByJsonUtils;
import com.winning.platwebservice.utils.XmlUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.dom4j.Document;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import javax.jws.WebService;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;


/**
 * 接口实现
 *
 * @author sj
 */
@WebService(serviceName = "CommonService", // 与接口中指定的name一致
        targetNamespace = "http://webservice.winning.com/" // 与接口中的命名空间一致,一般是接口的包名倒
)
@Service
public class CommonServiceImp implements CommonService {

    private static final Logger logger = LogManager.getLogger(CommonServiceImp.class);
//    final private String sqlServer = "Microsoft SQL Server";
//    final private String oracle = "Oracle";


    @Autowired
    DatabaseMetaData md;

    @Autowired
    EhcacheUtil ehcacheUtil;

    @Autowired
    DataProcessingDao receiveDao;

    @Autowired
    ValidateByJsonUtils validateUtils;

    @Autowired
    private KafkaCollect kafkaCollect;

    @Value("${kafka.collect.enable}")
    private boolean enable;

    @Override
    @Transactional
    public String DataService(String paramsXml) {

        String sResponse = null, code = null, massage = null;
//        ehcacheUtil.
        try {
            Document document = DocumentHelper.parseText(paramsXml);

            Element root = document.getRootElement();
            Element body = root.element("body");
            List<Element> childElements = body.elements();


            //得到表名
            String tableNmae = XmlUtils.getTableName(root);
            //得到字段名和字段值的Map集合
            List<Map> maps = XmlUtils.xml2List(childElements);

            //对入参进行校验
            Map<String, Object> restMap = batchCheckData(tableNmae, maps);

            if (restMap.get("res") != null && restMap.get("res").equals("0")) {


                return "<response>\n" +
                        "<code>" + "400" + "</code>\n" +
                        "<message>" + restMap.get("msg") + "</message>\n" +
                        "</response>\n";
            }


            //得到主键列表
            Set<String> primaryKeys = getPrimaryKey(tableNmae);





            for (Map map : maps) {
                /*进行一次日期格式转换*/
                map = ValidateByJsonUtils.mapDateConversion(map);

                Map<String, Object> primaryKey = new LinkedHashMap();
                for (String str : primaryKeys) {
                    primaryKey.put(str, map.get(str));
                }
                if (enable) {
                    Map sendData = new HashMap();
                    sendData.put("tableNmae", tableNmae);
                    sendData.put("primaryKey", primaryKey);
                    sendData.put("data", map);
                    kafkaCollect.producer(sendData);
                } else {
                    receiveDao.deleteData(tableNmae, primaryKey);

                    receiveDao.insertORupdate(tableNmae, map);
                }


            }


            code = "200";
            massage = "操作成功";

        } catch (Exception e) {
            logger.error("出现异常,事务回滚!!!\n" + e.getMessage());
            //事务回滚
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            code = "500";
            massage = e.getMessage();

        }
        sResponse = "<response>\n" +
                "<code>" + code + "</code>\n" +
                "<message>" + massage + "</message>\n" +
                "</response>\n";

        return sResponse;

    }

    /**
     * 数据规范性校验
     **/
    private Map<String, Object> batchCheckData(String bm, List<Map> maps) {
        for (Map map : maps) {
            Map restMap = validateUtils.checkData(map, bm);
            if (restMap.get("res").equals("0")) {
                return restMap;
            }
        }
        Map<String, Object> map = new LinkedHashMap<String, Object>();
        map.put("code", 200);
        map.put("message", "操作成功");
        return map;
    }

    /**
     * 根据表名得到表主键
     *
     * @param
     * @return
     * @throws
     */
    private Set<String> getPrimaryKey(String tableNmae) throws SQLException {

        ResultSet pkRset = null;
        try {
            pkRset = md.getPrimaryKeys(null, null, tableNmae);
            Set<String> pkName = new HashSet<String>();

            // 获取表主键名称列表
            while (pkRset.next()) {
                pkName.add(pkRset.getString("COLUMN_NAME"));
            }
            return pkName;
        } catch (SQLException e) {
            logger.error("获取{}主键列表失败", tableNmae, e);
            throw new RuntimeException(e);
        } finally {
            if (pkRset != null) {
                pkRset.close();
            }
        }


    }

}


